package org.luckyjourney.service.impl;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.luckyjourney.constant.RedisConstant;
import org.luckyjourney.entity.video.Video;
import org.luckyjourney.entity.user.User;
import org.luckyjourney.entity.vo.HotVideo;
import org.luckyjourney.entity.vo.Model;
import org.luckyjourney.entity.vo.UserModel;
import org.luckyjourney.service.InterestPushService;
import org.luckyjourney.service.video.TypeService;
import org.luckyjourney.util.RedisCacheUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.util.ObjectUtils;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

/**
 * @description:
 * @Author: Fxy
 * @CreateTime: 2023-10-26 11:54
 */
// 暂时为异步
@Service
public class InterestPushServiceImpl implements InterestPushService {

    @Autowired
    private RedisCacheUtil redisCacheUtil;

    @Autowired
    private TypeService typeService;

    @Autowired
    private RedisTemplate redisTemplate;


    final ObjectMapper objectMapper = new ObjectMapper();

    @Override
    @Async
    public void pushSystemStockIn(Video video) {
        // 视频添加到系统库中，并根据视频的标签将视频 ID 添加到相应的 Redis 集合中
        final List<String> labels = video.buildLabel();//方法返回视频的标签列表。
        final Long videoId = video.getId();// pipeline 特性，可以一次性执行多个 Redis 命令，提高了性能。
        redisTemplate.executePipelined((RedisCallback<Object>) connection -> {
            for (String label : labels) {
                connection.sAdd((RedisConstant.SYSTEM_STOCK + label).getBytes(), String.valueOf(videoId).getBytes());
            }
            return null;
        });
    }

    @Override
    @Async
    public void pushSystemTypeStockIn(Video video) {
        final Long typeId = video.getTypeId();
        redisCacheUtil.sSet(RedisConstant.SYSTEM_TYPE_STOCK + typeId, video.getId());
    }

    @Override
    public Collection<Long> listVideoIdByTypeId(Long typeId) {
        // 随机推送10个
        final List<Object> list = redisTemplate.opsForSet().randomMembers(RedisConstant.SYSTEM_TYPE_STOCK + typeId, 12);
        // 可能会有null
        final HashSet<Long> result = new HashSet<>();
        for (Object aLong : list) {
            if (aLong != null) {
                result.add(Long.parseLong(aLong.toString()));
            }
        }
        return result;
    }

    @Override
    @Async
    public void deleteSystemStockIn(Video video) {
        final List<String> labels = video.buildLabel();
        final Long videoId = video.getId();
        redisTemplate.executePipelined((RedisCallback<Object>) connection -> {
            for (String label : labels) {
                connection.sRem((RedisConstant.SYSTEM_STOCK + label).getBytes(), String.valueOf(videoId).getBytes());
            }
            return null;
        });
    }

    @Override
    @Async
    public void initUserModel(Long userId, List<String> labels) {

        final String key = RedisConstant.USER_MODEL + userId;
        Map<Object, Object> modelMap = new HashMap<>();
        if (!ObjectUtils.isEmpty(labels)) {
            final int size = labels.size();
            // 将标签分为等分概率,不可能超过100个分类
            double probabilityValue = 100 / size;
            for (String labelName : labels) {
                modelMap.put(labelName, probabilityValue);
            }
        }
        redisCacheUtil.del(key);
        redisCacheUtil.hmset(key, modelMap);
        // 为用户模型设置ttl TODO

    }


    @Override
    @Async
    public void updateUserModel(UserModel userModel) {

        final Long userId = userModel.getUserId();
        // 游客不用管
        if (userId != null) {
            final List<Model> models = userModel.getModels();
            // 获取用户模型
            String key = RedisConstant.USER_MODEL + userId;
            Map<Object, Object> modelMap = redisCacheUtil.hmget(key);

            if (modelMap == null) {
                modelMap = new HashMap<>();
            }
            for (Model model : models) {
                // 修改用户模型
                if (modelMap.containsKey(model.getLabel())) {
                    modelMap.put(model.getLabel(), Double.parseDouble(modelMap.get(model.getLabel()).toString()) + model.getScore());
                    final Object o = modelMap.get(model.getLabel());//获得概率值
                    if (o == null || Double.parseDouble(o.toString()) <= 0.0) {
                        modelMap.remove(model.getLabel());
                    }
                } else {
                    modelMap.put(model.getLabel(), model.getScore());
                }
            }

            // 每个标签概率同等加上标签数，再同等除以标签数  防止数据膨胀
            final int labelSize = modelMap.keySet().size();
            for (Object o : modelMap.keySet()) {
                modelMap.put(o, (Double.parseDouble(modelMap.get(o).toString()) + labelSize) / labelSize);
            }
            // 更新用户模型
            redisCacheUtil.hmset(key, modelMap);
        }

    }

    /**
     * 兴趣推送
     */
    @Override
    public Collection<Long> listVideoIdByUserModel(User user) {
        // 创建结果集
        Set<Long> videoIds = new HashSet<>(10);

        if (user != null) {
            final Long userId = user.getId();
            // 从模型中拿概率
            final Map<Object, Object> modelMap = redisCacheUtil.hmget(RedisConstant.USER_MODEL + userId);
            if (!ObjectUtils.isEmpty(modelMap)) {
                // 组成数组
                final String[] probabilityArray = initProbabilityArray(modelMap);
                final Boolean sex = user.getSex();
                // 获取视频
                final Random randomObject = new Random();
                final ArrayList<String> labelNames = new ArrayList<>();
                // 随机获取X个视频
                for (int i = 0; i < 8; i++) {
                    String labelName = probabilityArray[randomObject.nextInt(probabilityArray.length)];
                    labelNames.add(labelName);
                }
                // 提升性能 接口常量转化为局部变量
                String t = RedisConstant.SYSTEM_STOCK;
                // 随机获取
                List<Object> list = redisTemplate.executePipelined((RedisCallback<Object>) connection -> {
                    for (String labelName : labelNames) {
                        String key = t + labelName;
                        connection.sRandMember(key.getBytes());//从标签id库里随机拿视频
                    }
                    return null;
                });
                // 获取到的videoIds  过滤null值，将包含对象的列表转化成为一个包含长整型（Long）的集合（Set（不重复）
                Set<Long> ids = list.stream().filter(id -> id != null).map(id -> Long.parseLong(id.toString())).collect(Collectors.toSet());
                String key2 = RedisConstant.HISTORY_VIDEO;

                // 去重 推荐视频：滚动播放 推荐的视频少了 会有个预加载形势 可以理解为集合的扩容机制 到达了某个预值之后去拿数据
                /**
                 * : 最终结果存储在 simpIds 列表中。该列表包含了与 ids 中的每个 id 对应的 Redis 值。
                 * 如果某个 id 在 Redis 中没有对应的值，那么相应的列表元素将是 null。
                 */
                List simpIds = redisTemplate.executePipelined((RedisCallback<Object>) connection -> {
                    for (Long id : ids) {//.getBytes() 方法将键转换为字节数组，因为 Redis 的键通常是字节形式的。
                        connection.get((key2 + id + ":" + userId).getBytes());
                    }
                    return null;//在管道回调中返回 null。实际的返回值将由管道操作处理，并作为 simpIds 列表返回。
                });
                simpIds = (List) simpIds.stream().filter(o -> !ObjectUtils.isEmpty(o)).collect(Collectors.toList());
                ;
                if (!ObjectUtils.isEmpty(simpIds)) {
                    for (Object simpId : simpIds) {
                        final Long l = Long.valueOf(simpId.toString());
                        if (ids.contains(l)) {
                            ids.remove(l);
                        }
                    }
                }
                videoIds.addAll(ids);
                // 随机挑选一个视频,根据性别: 男：颜值 女：萌宠
                final Long aLong = randomVideoId(sex);
                if (aLong != null) {
                    videoIds.add(aLong);
                }
                return videoIds;
            }
        }
        // 游客
        // 随机获取10个标签
        final List<String> labels = typeService.random10Labels();
        final ArrayList<String> labelNames = new ArrayList<>();
        int size = labels.size();
        final Random random = new Random();
        // 获取随机的标签
        if (size > 0) {
            for (int i = 0; i < 10; i++) {
                final int randomIndex = random.nextInt(size);
                labelNames.add(RedisConstant.SYSTEM_STOCK + labels.get(randomIndex));
            }
        } else {
            // 处理 size 为 0 的情况，比如给出默认值或者抛出异常
            throw new IllegalArgumentException("Labels标签 list is empty");
        }
        // 获取videoId
        final List<Object> list = redisCacheUtil.sRandom(labelNames);
        if (!ObjectUtils.isEmpty(list)) {
            videoIds = list.stream().filter(id -> !ObjectUtils.isEmpty(id)).map(id -> Long.valueOf(id.toString())).collect(Collectors.toSet());
        }

        return videoIds;
    }

    /**
     * 根据标签获取视频id
     */
    @Override
    public Collection<Long> listVideoIdByLabels(List<String> labelNames) {
        final ArrayList<String> labelKeys = new ArrayList<>();
        for (String labelName : labelNames) {
            labelKeys.add(RedisConstant.SYSTEM_STOCK + labelName);
        }
        Set<Long> videoIds = new HashSet<>();
        final List<Object> list = redisCacheUtil.sRandom(labelKeys);
        if (!ObjectUtils.isEmpty(list)) {
            videoIds = list.stream().filter(id -> !ObjectUtils.isEmpty(id)).map(id -> Long.valueOf(id.toString())).collect(Collectors.toSet());
        }
        return videoIds;
    }

    @Override
    @Async
    public void deleteSystemTypeStockIn(Video video) {
        final Long typeId = video.getTypeId();
        redisCacheUtil.setRemove(RedisConstant.SYSTEM_TYPE_STOCK + typeId, video.getId());
    }


    public Long randomHotVideoId() {
        final Object o = redisTemplate.opsForZSet().randomMember(RedisConstant.HOT_RANK);
        try {
            return objectMapper.readValue(o.toString(), HotVideo.class).getVideoId();
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        }
        return null;
    }

    public Long randomVideoId(Boolean sex) {
        String key = RedisConstant.SYSTEM_STOCK + (sex ? "颜值" : "萌宠");
        final Object o = redisCacheUtil.sRandom(key);
        if (o != null) {
            return Long.parseLong(o.toString());
        }
        return null;
    }

    // 随机获取视频id
    public Long getVideoId(Random random, String[] probabilityArray) {
        String labelName = probabilityArray[random.nextInt(probabilityArray.length)];
        // 获取对应所有视频
        String key = RedisConstant.SYSTEM_STOCK + labelName;
        final Object o = redisCacheUtil.sRandom(key);
        if (o != null) {
            return Long.parseLong(o.toString());
        }
        return null;
    }

    // 初始化概率数组 -> 保存的元素是标签  每个标签根据其在modelMap中对应的概率值出现相应次数
    public String[] initProbabilityArray(Map<Object, Object> modelMap) {
        // key: 标签  value：概率
        Map<String, Integer> probabilityMap = new HashMap<>();
        int size = modelMap.size();
        //原子操作的类，它提供了一种线程安全的方式来操作整数值。
        final AtomicInteger n = new AtomicInteger(0);
        modelMap.forEach((k, v) -> {
            // 防止结果为0,每个同等加上标签数
            int probability = (((Double) v).intValue() + size) / size;
            n.getAndAdd(probability); //概率值累加
            probabilityMap.put(k.toString(), probability);
        });
        //根据累加后的概率值 n 创建了一个字符串数组 probabilityArray，其长度为 n.get()累加概率值
        final String[] probabilityArray = new String[n.get()];

        final AtomicInteger index = new AtomicInteger(0);
        // 初始化数组 根据每个标签的概率值，将标签填充到probabilityArray中相应的位置。
        probabilityMap.forEach((labelsId, p) -> {
            int i = index.get();
            int limit = i + p;
            while (i < limit) {
                probabilityArray[i++] = labelsId;
            }
            index.set(limit);
        });
        return probabilityArray;
    }


}
